package com.wgyscsf.rx2demo;

import android.os.Bundle;
import android.util.Log;
import android.view.View;
import android.widget.Button;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

public class MainActivity extends BaseActivity {

    Button oftenUseBtn;
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
    }

    public void oftenUse(View view) {
        oftenUseBtn= (Button) view;
        oftenUseRx();

    }

    private void oftenUseRx() {
        Observable<String> lStringObservable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                Log.e(TAG, "subscribe: "+","+Thread.currentThread().getName() );
                e.onNext("abc");
                e.onNext("def");
                e.onNext("ghi");
                //e.onComplete();
            }
        });

        Observer<String> lStringObserver1 = new Observer<String>() {

            @Override
            public void onSubscribe(Disposable d) {
                addDisposable(d);//在页面销毁时，即使取消请求
                Log.e(TAG, "onSubscribe1: " + d.isDisposed() + "," + Thread.currentThread().getName());
            }

            @Override
            public void onNext(String value) {
                Log.e(TAG, "onNext1: " + value + "," + Thread.currentThread().getName());
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: " + e.getMessage() + "," + Thread.currentThread().getName());
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete1: " + "," + Thread.currentThread().getName());
            }
        };

        Observer<String> lStringObserver2 = new Observer<String>() {

            @Override
            public void onSubscribe(Disposable d) {
                addDisposable(d);//在页面销毁时，即使取消请求
                Log.e(TAG, "onSubscribe2: " + d.isDisposed() + "," + Thread.currentThread().getName());
            }

            @Override
            public void onNext(String value) {
                Log.e(TAG, "onNext2: " + value + "," + Thread.currentThread().getName());
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError2: " + e.getMessage() + "," + Thread.currentThread().getName());
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete2: " + "," + Thread.currentThread().getName());
            }
        };

        lStringObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());

        lStringObservable.subscribe(lStringObserver1);
        lStringObservable.subscribe(lStringObserver2);
    }

    public void simpleUse(View view) {
        simpleUse();
    }

    private void simpleUse() {
        //完整的链式调用
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                Log.e(TAG, "subscribe: "+ "," + Thread.currentThread().getName() );
                e.onNext(1);
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException pE) {
                    pE.printStackTrace();
                }
                e.onNext(2);
                e.onComplete();
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer pInteger) throws Exception {
                Log.e(TAG, "accept(Integer pInteger): "+pInteger+ "," + Thread.currentThread().getName() );
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable pThrowable) throws Exception {
                Log.e(TAG, "accept(Throwable pThrowable): "+pThrowable+ "," + Thread.currentThread().getName() );
            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                Log.e(TAG, "run: "+ "," + Thread.currentThread().getName() );
            }
        }, new Consumer<Disposable>() {
            @Override
            public void accept(Disposable pDisposable) throws Exception {
                Log.e(TAG, "accept: 加入取消队列" );
                addDisposable(pDisposable);//在页面销毁时，即使取消请求
                Log.e(TAG, "accept(Disposable pDisposable): "+pDisposable.isDisposed() );
            }
        });

        //简洁版，只需要接受onNext方法
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(6666);
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer pInteger) throws Exception {
                        Log.e(TAG, "accept: "+pInteger );
                    }
                });
    }

    public void mapUse(View view) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(6666);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer pInteger) throws Exception {
                String preStr="待处理字符串";

                return preStr+pInteger;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String pS) throws Exception {
                Log.e(TAG, "accept: 处理之后的字符串："+pS );
            }
        });
    }

    public void flatMap(View view) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                try {
                    Thread.sleep(300);
                } catch (InterruptedException pE) {
                    pE.printStackTrace();
                }
                e.onNext(2);
                try {
                    Thread.sleep(300);
                } catch (InterruptedException pE) {
                    pE.printStackTrace();
                }
                e.onNext(3);
                Log.e(TAG, "subscribe: "+Thread.currentThread().getName());
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())//切换到主线程
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer pInteger) throws Exception {
                        Log.e(TAG, "accept: "+pInteger+","+Thread.currentThread().getName() );
                    }
                }).observeOn(Schedulers.io())//切换到io线程，处理拼接字符串
                .flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(final Integer pInteger) throws Exception {
                Log.e(TAG, "applyapply(final Integer pInteger): "+Thread.currentThread().getName());

                final String preStr="我是转换之后的字符串，可以将上游的int类型转为为string";
                try {
                    Thread.sleep(300);
                } catch (InterruptedException pE) {
                    pE.printStackTrace();
                }
                return Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> e) throws Exception {
                        e.onNext(preStr+pInteger);
                        try {
                            Thread.sleep(300);
                        } catch (InterruptedException pE) {
                            pE.printStackTrace();
                        }
                        Log.e(TAG, "subscribe(ObservableEmitter<String> e): "+Thread.currentThread().getName());

                    }
                });
            }
        }).observeOn(AndroidSchedulers.mainThread())//切换到主线程进行显示
                .subscribe(new Consumer<String>() {
            @Override
            public void accept(String pS) throws Exception {
                Log.e(TAG, "accept: "+pS+","+Thread.currentThread().getName() );
            }
        });

    }

    public void zip(View view) {
        //处理用户id
        Observable<Integer> lIntegerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).subscribeOn(Schedulers.io());

        //处理用户名字
        Observable<String> lStringObservable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("张三");
                e.onNext("李四");
                Log.e(TAG, "subscribe: "+","+Thread.currentThread().getName() );
            }
        }).subscribeOn(Schedulers.io());

        //组合用户信息
        Observable<User> lUserObservable = Observable.zip(lIntegerObservable, lStringObservable, new BiFunction<Integer, String, User>() {
            @Override
            public User apply(Integer pInteger, String pS) throws Exception {
                User lUser = new User(pInteger, pS);
                Log.e(TAG, "apply: "+","+Thread.currentThread().getName() );
                return lUser;
            }
        }).subscribeOn(Schedulers.io());

        //在主线程接收处理后的信息
        lUserObservable.observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<User>() {
            @Override
            public void accept(User pUser) throws Exception {
                Log.e(TAG, "accept: "+pUser.toString()+","+Thread.currentThread().getName() );
            }
        });
    }

    static class User{
        private int id;
        private String name;

        User(int id,String name){
            this.id=id;
            this.name=name;
        }

        public int getId() {
            return id;
        }

        public void setId(int pId) {
            id = pId;
        }

        public String getName() {
            return name;
        }

        public void setName(String pName) {
            name = pName;
        }

        @Override
        public String toString() {
            return "User{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    '}';
        }
    }
}
